package com.my.serialize;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.TimeZone;
import java.util.stream.Collectors;


public class MySqlDeserialization implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        //返回结果
        JSONObject result = new JSONObject();
        //表主键
        Struct primaryKey = (Struct) sourceRecord.key();
        Schema primaryKeySchema = primaryKey.schema();
        List<String> primaryKeyList = primaryKeySchema.fields().stream().map(Field::name).collect(Collectors.toList());
        result.put("primaryKey", primaryKeyList);
        //库名表名
        String[] split = sourceRecord.topic().split("\\.");
        if (split.length == 1){
            return;
        }
        String dataBase = split[1];
        String tableName = split[2];
        //值
        Struct value = (Struct) sourceRecord.value();
        //before
        Struct before = value.getStruct("before");
        JSONObject beforeJson = new JSONObject();
        if (before != null) {
            Schema schema = before.schema();
            for (Field field : schema.fields()) {
                Object fieldValue = before.get(field);
                if (fieldValue != null) {
                    beforeJson.put(field.name(), parseValue(field, fieldValue));
                } else {
                    beforeJson.put(field.name(), "");
                }
            }
        }
        //after
        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            Schema schema = after.schema();
            for (Field field : schema.fields()) {
                Object fieldValue = after.get(field);
                if (fieldValue != null) {
                    afterJson.put(field.name(), parseValue(field, fieldValue));
                } else {
                    afterJson.put(field.name(), "");
                }
            }
        }
        //操作
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String opType = operation.toString().toUpperCase();
        //组装
        result.put("database", dataBase);
        result.put("tableName", tableName);
        result.put("before", beforeJson);
        result.put("after", afterJson);
        result.put("opType", opType);

        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

    private Object parseValue(Field field, Object fieldValue) throws ParseException {
        if (field != null && fieldValue != null) {
            if ("int32".equals(field.schema().type().getName())
                    && "io.debezium.time.Date".equals(field.schema().name())) {
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                String beforeValue2Strs = String.valueOf(fieldValue);
                long times = Long.parseLong(beforeValue2Strs) * 24 * 60 * 60 * 1000L;
                String date = sdf.format(new Date(times));
                return date;
            } else if ("int64".equals(field.schema().type().getName())
                    && "io.debezium.time.Timestamp".equals(field.schema().name())) {
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String beforeValue2Str = String.valueOf(fieldValue);
                long times = Long.parseLong(beforeValue2Str)-(long)(8 *60 * 60 * 1000);
                String dataTime = sdf.format(times);
                return dataTime;
            }else if ("string".equals(field.schema().type().getName())
                    && "io.debezium.time.ZonedTimestamp".equals(field.schema().name())) {
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                sdf.setTimeZone(TimeZone.getTimeZone("UTC" ));
                String utcTime = String.valueOf(fieldValue);
                Date date = sdf.parse(utcTime);
                sdf.applyPattern("yyyy-MM-dd HH:mm.ss");
                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8:00"));
                String zonedTime = sdf.format(date);
                return zonedTime;
            }else{
                return fieldValue;
            }
        }
        return "";
    }

}
